package cn.sheep.dolphin.graphxlearning

import cn.sheep.dolphin.common.DolphinAppComm
import org.apache.log4j.{Level, Logger}
import org.apache.spark.graphx.{Edge, Graph, VertexId}
import org.apache.spark.rdd.RDD

/**
  * author: old sheep
  * QQ: 64341393 
  * Created 2018/12/3
  */
object GoodFriends {

	Logger.getLogger("org").setLevel(Level.WARN)

	def main(args: Array[String]): Unit = {
		val sc = DolphinAppComm.createSparkContext("找对象")

		// 构建点集合 RDD[(Long, VD)]
		val vertex: RDD[(VertexId, (String, Int))] = sc.makeRDD(Seq(
			(1L, ("zhangsan", 18)),
			(2L, ("xiaoming", 18)),
			(9L, ("laowang", 18)),
			(6L, ("wukong", 18)),
			(133L, ("bajie", 18)),

			(16L, ("wujingjing", 18)),
			(44L, ("baijingjing", 18)),
			(21L, ("zixia", 18)),
			(138L, ("xiaotiantian", 18)),

			(5L, ("niumowang", 18)),
			(7L, ("tieshangongzhu", 18)),
			(158L, ("qinxianglin", 18))
		))

		// 构建边集合 RDD[Edge[ED]]
		val edges: RDD[Edge[String]] = sc.makeRDD(Seq(
			Edge(1, 133, ""),
			Edge(133, 2, ""),
			Edge(6, 133, ""),
			Edge(9, 133, ""),
			Edge(6, 138, ""),
			Edge(16, 138, ""),
			Edge(21, 138, ""),
			Edge(44, 138, ""),
			Edge(5, 158, ""),
			Edge(7, 158, "")
		))

		// 图对象 = 点集合 + 边集合
		val graph = Graph(vertex, edges)
		// 调用连通图算法，找图中的可以联通的分支
		val cc = graph.connectedComponents()
			// 让处在同一个连通图分支中的所有点都向该分支中最小的点进行靠拢，组合成元组形式 （自己, 自己所处的分支图中最小的点ID）
			.vertices

		cc.join(vertex).map{
			case (cmId, (minId, (name, age))) => (minId, List((name, age)))
		}.reduceByKey(_ ++ _).foreach(println)

		// val result = cc.map(tp => (tp._2, List(tp._1))).reduceByKey(_ ++ _)

		sc.stop()
	}

}
